//********************************************************************************
//
// File :           ph02_enum_sort_cluster.cpp
//
// Description :    enum_sort arrays using cluster
//
// Dependencies:    mpi.h
//
// Author :         UlMonkey1987( WU Kan )
//
// Stu NO. :        2010212404
//
// Date :           2011-4-17
//
//********************************************************************************

#include "mpi.h"

#include <iostream>
#include <algorithm>
#include <cmath>
#include <iomanip>
#include <fstream>
#include <cstdlib>

#include <vector>
#include <string>

using namespace std;

// bool readData( const string& strTestDataPath, long* pNum, const unsigned long numN )
bool readData( const string& strTestDataPath, long* pNum, const unsigned long numN )
{
    if ( 0 == pNum ) return false;

    ifstream fin( strTestDataPath.c_str(), iostream::binary );

    char* pByteData = ( char* ) pNum;

    fin.read( pByteData, numN * sizeof( long ) );

    return true;
}
//

// long getNthElem( long* pNum, const unsigned long numN, const unsigned long nthIndex )
long getNthElem( long* pNum, const unsigned long numN, const unsigned long nthIndex )
{
    nth_element( pNum, pNum + nthIndex, pNum + numN );

    return pNum[nthIndex];
}
//

// unsigned long getBucketIndex( const long n, const long procN, long* pThreshold )
unsigned long getBucketIndex( const long n, const long procN, long* pThreshold )
{
    for ( long i = 0; i < procN; i ++ )
    {
        if ( n <= pThreshold[i] ) return i;
    }

    return procN - 1;
}
//

// bool insideSort( long* pNum, const unsigned long startIndex, const unsigned long endIndex )
bool insideSort( long* pNum, const unsigned long startIndex, const unsigned long endIndex )
{
    sort( pNum + startIndex, pNum + endIndex + 1 );

    return true;
}
//

int main( int argc, char* argv[] )
{
    if ( argc < 4 )
    {
        cout << "error : lack args." << endl;
        cout << "usage : " << "<program_name>" << " <test_data_path> <n> <output_file_path>." << endl;
        return 1;
    }

    string strTestDataPath = string( argv[1] );     // data src path
    unsigned long numN = atoi( argv[2] );      // numN
    string strOutPath = string( argv[3] );      // output path

//     // args for test
//     string strTestDataPath = "";
//     unsigned long numN = 100;
//     string strOutPath = "result.txt";
//     // ! args for test

    const int MAX_PROC_NAME_LEN = 256;      // max length of a processor name

    int procN;                           // number of process
    int curProcId;                          // id of current process
    char curProcName[MAX_PROC_NAME_LEN];      // name of current process
    int curProcNameLen;                     // the length of current process name

    MPI_Init( &argc, &argv );
    MPI_Comm_size( MPI_COMM_WORLD, &procN );
    MPI_Comm_rank( MPI_COMM_WORLD, &curProcId );
    MPI_Get_processor_name( curProcName, &curProcNameLen );

    cout << "Process " << curProcId << " Running..." << endl << endl;

    double timeStart, timeEnd, timeCost;        // start time, end time, duration time

    long* pNum = 0;                             // data to be sorted
    long* pNumTemp = 0;                         // data in the local bucket
    int* pNumN = new int[procN];                    // data count for each process
    int* pOffset = new int[procN];                  // data offset for each process

    if ( 0 == curProcId )
    {
        pNum = new long[numN];
        readData( strTestDataPath, pNum, numN );
//         for ( unsigned long i = 0; i < numN; i ++ )
//             pNum[i] = numN - i;

        // separate pNum into different bucket
        long delta = numN / procN;

        long* pThreshold = new long[procN];
        for ( long i = 0; i < procN; i ++ )
        {
            unsigned long tIndex = delta * ( i + 1 );
            pThreshold[i] = getNthElem( pNum, numN, tIndex );       // select n-th elem
        }

        pNumN = new int[procN];
        for ( long i = 0; i < procN; i ++ )
            pNumN[i] = 0;

        vector< vector< long > > vecSubNum( procN );
        for ( unsigned long i = 0; i < numN; i ++ )
        {
            unsigned long bucketIndex = getBucketIndex( pNum[i], procN, pThreshold );

            vecSubNum[bucketIndex].push_back( pNum[i] );
            pNumN[bucketIndex] ++;
        }

        delete []pThreshold; pThreshold = 0;

        unsigned long tIndex = 0;
        for ( unsigned long i = 0; i < vecSubNum.size(); i ++ )
        {
            for ( unsigned long j = 0; j < vecSubNum[i].size(); j ++ )
            {
                pNum[tIndex] = vecSubNum[i][j]; tIndex ++;
            }
        }

        pOffset[0] = 0;
        for ( long i = 1; i < procN; i ++ )
           pOffset[i] = pOffset[i-1] + pNumN[i-1];

        // timer started
        timeStart = MPI_Wtime();
        // ! timer started
    }

    MPI_Bcast( pNumN, procN, MPI_INT, 0, MPI_COMM_WORLD );          // data count for each process
    MPI_Bcast( pOffset, procN, MPI_INT, 0, MPI_COMM_WORLD );        // data offset for each process

    int curNumN = pNumN[curProcId];                        // data count on current process

    pNumTemp = new long[curNumN];

    // seperate data into different buckets
    MPI_Scatterv( pNum, pNumN, pOffset, MPI_LONG, pNumTemp, curNumN, MPI_LONG, 0, MPI_COMM_WORLD );
    // ! seperate data into different buckets

    // sort data inside current bucket
    insideSort( pNumTemp, 0, curNumN - 1 );
    // ! sort data inside current bucket

    // gather sorted data from different buckets
    MPI_Gatherv( pNumTemp, curNumN, MPI_LONG, pNum, pNumN, pOffset, MPI_LONG, 0, MPI_COMM_WORLD );
    // ! gather sorted data from different buckets

    if ( 0 == curProcId )
    {
        // timer stopped
        timeEnd = MPI_Wtime();
        // ! timer stopped

        timeCost = timeEnd - timeStart;

//         for ( unsigned long i = 0; i < numN; i ++ )
//         {
//             cout << pNum[i];
//             if ( numN - 1 != i ) cout << ",";
//         }
//         cout << endl;
// 
//         cout << endl << "cost time : " << timeCost << endl;

        delete []pNum; pNum = 0;

        // output result
        ofstream fout( strOutPath.c_str(), ios::app );
        fout << "cluster\t" << strTestDataPath << "\t" << "bucket_sorting\t" << procN << "\t" << numN << "\t" << setprecision( 2 ) << timeCost << "\n";
        // ! output result

        cout << endl << "cluster " << strTestDataPath << " " << "bucket_sorting " << procN << " " << numN << " " << setprecision( 2 ) << timeCost << endl;
    }

    delete []pNumTemp; pNumTemp = 0;
    delete []pNumN; pNumN = 0;
    delete []pOffset; pOffset = 0;

    // system( "pause" );

    MPI_Finalize();

    return 0;
}
